package com.tianji.aigc.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.IdUtil;
import com.tianji.aigc.constant.ToolConstant;
import com.tianji.aigc.dto.ChatDto;
import com.tianji.aigc.entity.ChatSession;
import com.tianji.aigc.enums.ChatEventTypeEnum;
import com.tianji.aigc.memory.RedisChatMemory;
import com.tianji.aigc.properties.SessionProperties;
import com.tianji.aigc.service.IChatService;
import com.tianji.aigc.service.IChatSessionService;
import com.tianji.aigc.tools.ToolResultHolder;
import com.tianji.aigc.vo.ChatEventVO;
import com.tianji.common.exceptions.BadRequestException;
import com.tianji.common.utils.UserContext;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.client.advisor.AbstractChatMemoryAdvisor;
import org.springframework.ai.chat.client.advisor.QuestionAnswerAdvisor;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.ai.vectorstore.SearchRequest;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 对话业层实现类
 *
 * @Author mr.wu
 * @Date 2025-9-2 16:06
 */
@Service
public class ChatServiceImpl implements IChatService {

    @Autowired
    private ChatClient chatClient;

    @Autowired
    private IChatSessionService chatSessionService;

    /**
     * 定义可以控制连续输出标识值的Map，Map的KEY是sessionId，Vlue就是布尔值，true表示可以继续输出，false表示中断输出
     */
    private static Map<String,Boolean> CONTINUE_FLAG_MAP = new ConcurrentHashMap<>();

    @Autowired
    private VectorStore vectorStore;

    /**
     * 流式对话
     *
     * @param dto
     * @return
     */
    @Override
    public Flux<ChatEventVO> chat(ChatDto dto) {
        String sessionId = dto.getSessionId();
        //判断会话是否存在
        Long count = chatSessionService.lambdaQuery().eq(ChatSession::getSessionId, sessionId).count();
        if(count==0) {
            throw new BadRequestException("请先创建会话");
        }

        //保存会话对应的第一次对话的标题信息
        chatSessionService.updateSessionTitle(UserContext.getUser(),sessionId, dto.getQuestion());

        //拼接对话ID： 用户ID+会话ID
        String conversationId = UserContext.getUser() + ":" + sessionId;

        StringBuffer content = new StringBuffer();//线程安全的字符串拼接类，用于拼接流式输出的每个字符串

        String requestId = IdUtil.fastSimpleUUID();//请求ID，唯一值

        ChatEventVO STOP_EVENT_DATA = ChatEventVO.builder().eventType(ChatEventTypeEnum.STOP.getValue()).build();

        //进行大模型对话
        return chatClient.prompt()
                .user(dto.getQuestion())
                .toolContext(MapUtil.<String,Object>builder()
                        .put(ToolConstant.REQUEST_ID, requestId)
                        .put(ToolConstant.USER_ID,UserContext.getUser())
                        .build())//将请求ID设置到ToolCalling共享容器里
                .advisors(x->x
                        .advisors(new QuestionAnswerAdvisor(vectorStore, SearchRequest.builder().query("").topK(999).build()))
                        .param(AbstractChatMemoryAdvisor.CHAT_MEMORY_CONVERSATION_ID_KEY,conversationId))
                .stream()
                //.content()
                .chatResponse()
                .doFirst(()->{ CONTINUE_FLAG_MAP.put(sessionId,true); })//标记开始输出的第一步，设置为true，表示可以继续输出
                .doOnError(x->  CONTINUE_FLAG_MAP.remove(sessionId)) //标记输出出现异常，删除map，节省内存空间
                .doOnComplete(()->{ CONTINUE_FLAG_MAP.remove(sessionId); })//标记输出全部完成，删除map，节省内存空间
                .doOnCancel(()->{this.saveToRedis(conversationId, content.toString());}) //发现取消输出时（输出停止时）将之前收集到的全部消息保存到Redis
                .takeWhile(x->CONTINUE_FLAG_MAP.getOrDefault(sessionId, false)) //根据布尔值决定是否继续输出，true继续，false停止
                .map( x-> {
                            //content.append(x);
                    //如果是流式输出正常业务数据的最后一条消息,那么就要获取该消息的唯一ID
                    String finishReason = x.getResult().getMetadata().getFinishReason();
                    if (finishReason.equals("STOP")) {
                        String messageId = x.getMetadata().getId();//消息唯一ID

                        //将消息ID和请求ID的值的关系存入结果保持器中
                        ToolResultHolder.put(messageId,ToolConstant.REQUEST_ID,requestId);
                    }
                    String text = x.getResult().getOutput().getText();
                    //大模型流式输出返回的业务数据和类型为1001
                            //return ChatEventVO.builder().eventData(x).eventType(ChatEventTypeEnum.DATA.getValue()).build();
                            return ChatEventVO.builder().eventData(text).eventType(ChatEventTypeEnum.DATA.getValue()).build();
                        }
                )
                .concatWith(Flux.defer(()->{
                             //从结果保持器中，根据请求ID获取结果信息（可能是课程详情或订单等信息）
                             Map<String, Object> resultMap = ToolResultHolder.get(requestId);
                             if(CollUtil.isNotEmpty(resultMap)) {
                                 //如果结果信息非空，就输出1003参数事件数据给前端
                                 ChatEventVO chatEventVO = ChatEventVO.builder().eventData(resultMap).eventType(ChatEventTypeEnum.PARAM.getValue()).build();
                                 ToolResultHolder.remove(requestId);//用完就删，腾出内存空间
                                 return Flux.just(chatEventVO, STOP_EVENT_DATA);
                             }

                            return Flux.just(STOP_EVENT_DATA);
                        })
                );//输出结束指定类型为1002，必须指定这个，否则前端只能对话一次
    }

    @Autowired
    private RedisChatMemory redisChatMemory;

    private void saveToRedis(String conversationId, String content) {
        redisChatMemory.add(conversationId,new AssistantMessage(content));
    }

    /**
     * 停止生成
     *
     * @param sessionId
     */
    @Override
    public void stop(String sessionId) {
        //核心思路：打断stream流式输出，不让继续输出，必须改变可以连续输出的布尔标识的值，将值设置为false
        CONTINUE_FLAG_MAP.put(sessionId,false);
    }

    @Autowired
    private SessionProperties sessionProperties;

    /**
     * 文本聊天
     *
     * @param text
     * @return
     */
    @Override
    public String chatText(String text) {
        return chatClient.prompt()
                .system(sessionProperties.getText())
                .user(text)
                .call()
                .content();
    }
}
